/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl.disk;

import java.io.IOException;
import java.util.concurrent.locks.Lock;

import org.sabayframework.mem.BigBuffer;
import org.sabayframework.mem.Memorizable;
import org.sabayframework.mem.Transaction.Status;
import org.sabayframework.mem.cache.Cache;
import org.sabayframework.mem.impl.TransactionImpl;
import org.sabayframework.mem.impl.TxActivity;

public class NoLockTxActivity extends TxActivity {

	protected final DiskTransaction tx;
	private final DiskSpace space;
	private final Cache<Memorizable> rCache;
	private final Cache<Memorizable> wCache;
	
	
	public NoLockTxActivity(TransactionImpl transaction, DiskSpace tm) throws IOException {
		super(transaction, tm);
		tx = tm.getStorage().beginTransaction();
		this.space = tm; 
		this.rCache = space.getCacheFactory().createCache();
		this.wCache = space.getCacheFactory().createCache();
	}
	
	@Override
	protected Boolean[] contains(long[] ids) throws Exception {
		Boolean[] res = new Boolean[ids.length];
		for(int i=0;i<ids.length;++i) {
			final long id = ids[i];
			res[i] = Boolean.FALSE;
			if(wCache.get(id) == null && rCache.get(id)==null){
				byte[] data = tx.read(id);
				if(data == null || data.length==0) continue;
			}
			res[i] = Boolean.TRUE;
		}
		return res;
	}
	
	@Override
	protected Memorizable[] read(Class<? extends Memorizable> objectClass, long[] ids) throws Exception {
		final Cache<BigBuffer> spaceCache = space.getCache();
		Memorizable[] res= new Memorizable[ids.length]; 
		for(int i=0;i<ids.length;++i) {
			final long id = ids[i];
			Memorizable m = wCache.get(id);
			if(m == null) m = rCache.get(id);
			if(m == null){
				if(m == null && wCache.contains(id)) {
					m = objectClass.newInstance();
					m.setId(id);
				}else if(!tx.hasModified(id)){
					BigBuffer b = spaceCache.get(id);
					if(b != null){
						b = b.asReadOnlyBuffer();
						b.rewind();
						m = objectClass.newInstance();
						m.setId(id);
						m.deserialize(b);
						rCache.put(id, m);
					}
				}
			}
			if(m == null){
				final long lastCommitted = space.getLastCommittedId();
				byte[] data = tx.read(id);
				BigBuffer bb = null;
				if(data!=null && data.length>0) bb = space.wrap(data, 0, data.length);
				m = objectClass.newInstance();
				m.setId(id);
				if(bb!=null) m.deserialize(bb);
				//put modified to tx cache and unmodified to space cache
				boolean modified = tx.hasModified(id);
				if(modified) {
					wCache.put(id, m);
				} else {
					final Lock lock = space.getLock().writeLock();
					lock.lock();
					try{
						if(space.getLastCommittedId()!=lastCommitted){
							//another tx committed, check space cache
							BigBuffer b = spaceCache.get(id);
							if(b != null){
								b = b.asReadOnlyBuffer();
								b.rewind();
								m = objectClass.newInstance();
								m.setId(id);
								m.deserialize(b);
							}else{
								spaceCache.put(id, bb);
							}
						}
					}finally{
						lock.unlock();
					}
					rCache.put(id, m);
				}
			}
			res[i] = m;
		}
		return res;
	}

	@Override
	protected boolean write(Memorizable[] rs) throws Exception {
		for(Memorizable r:rs){
			long addr = r.getId();
			wCache.put(addr, r);
			rCache.remove(addr);
		}
		return true;
	}

	@Override
	protected boolean delete(long[] ids) throws Exception {
		for(int i=0;i<ids.length;++i) {
			long addr = ids[i];
			wCache.put(addr, null);
			rCache.remove(addr);
		}
		return true;
	}

	@Override
	protected boolean complete(Status status) throws Exception {
		if(status == Status.COMMIT) {
			//copy from wCache to disk
			final Cache<BigBuffer> spaceCache = space.getCache();
			final Lock lock = space.getLock().writeLock();
			lock.lock();
			try{
				wCache.iterateAll(new Cache.Accessor<Memorizable>() {
					private BigBuffer buffer;
					@Override
					public void access(long address, Memorizable data) throws Exception {
						if(data == null) {
							tx.delete(address);
							spaceCache.remove(address);
						}else{
							final int maxLength = data.getMaxBufferSize();
							if(buffer == null || buffer.capacity() < maxLength){
								buffer = getSpace().wrap(new byte[maxLength], 0, maxLength);
							}
							buffer.clear();
							data.serialize(buffer);
							buffer.flip();
							int serSize = (int)buffer.limit();
							byte[] bytes = new byte[serSize];
							buffer.get(bytes);
							tx.update(address, bytes);
							
							BigBuffer cached = space.wrap(bytes, 0, serSize);
							spaceCache.put(address, cached);
						}
					}
				});
			} finally{
				lock.unlock();
			}
			tx.commit();
		}
		else {
			tx.rollback();
		}
		wCache.clear();
		rCache.clear();
		return true;
	}

}
